package de.mmis.core.pojop.server;

import java.io.IOException;
import java.io.InputStream;
import java.io.OutputStream;
import java.net.ServerSocket;
import java.net.Socket;
import java.net.SocketException;

import org.apache.log4j.Logger;

import de.mmis.core.base.Toolbox;
import de.mmis.core.base.event.AbstractObservable;
import de.mmis.core.pojop.answer.Answer;
import de.mmis.core.pojop.processor.RequestProcessorFactory;
import de.mmis.core.pojop.processor.RequestProcessorFactory.RequestProcessor;
import de.mmis.core.pojop.request.Request;

/**
 * abstract implementation of the server interface providing basic functions
 * 
 * @author Sebastian Bader
 * @author Christian Koch
 */
public abstract class AbstractServer<R extends Request> extends
		AbstractObservable<Server.ConnectionEvent> implements Server {

	private final static Logger LOGGER = Logger.getLogger(AbstractServer.class);

	private final static ThreadGroup THREAD_GROUP = new ThreadGroup(
			"NetworkPublisher");
	private final static ThreadGroup CLIENT_THREAD_GROUP = new ThreadGroup(
			THREAD_GROUP, "NWP client threads");

	protected RequestProcessorFactory<R> requestProcessorFactory;
	private ServerSocket serverSocket; // listening socket
	private boolean keepListening; // server will stop listening for new client
									// connections if set to false

	private String simpleName = getClass().getSimpleName();

	/**
	 * creates new abstract server. it will not listening to clients until
	 * calling start()
	 * 
	 * @param port
	 *            port number server should listen to clients on
	 * @throws IOException
	 *             the creation of the server socket failed
	 */
	public AbstractServer(int port,
			RequestProcessorFactory<R> requestProcessorFactory)
			throws IOException {
		this.keepListening = true;
		this.serverSocket = new ServerSocket(port, 10);

		this.requestProcessorFactory = requestProcessorFactory;
	}

	/**
	 * writes a string to stdout (if not silent)
	 * 
	 * @param s
	 *            string to be written to stdout
	 */
	public void stdout(String s) {
		LOGGER.info(s);
	}

	@Override
	public Thread start(String name) {
		Thread thread = new Thread(THREAD_GROUP, this, name);
		thread.start();
		return thread;
	}

	/**
	 * creates a session handler for communication with newly connected client
	 * on given socket
	 * 
	 * @param socket
	 *            socket with new client connection
	 * @return a (not started) session handler for the new client
	 */
	public abstract SessionHandler createSessionHandler(Socket socket);

	@Override
	public void run() {
		while (this.keepListening) {
			// wait for and handle incoming connections
			try {
				stdout("Wait for incomming connections on "
						+ this.serverSocket.getLocalPort() + "...");
				Socket socket = this.serverSocket.accept();

				SessionHandler sessionHandler = createSessionHandler(socket);
				(new Thread(CLIENT_THREAD_GROUP, sessionHandler, simpleName
						+ " client " + socket.toString())).start();

			} catch (SocketException e) {
				if (!this.keepListening) {
					// we can simply ignore it if keepListening is set to false,
					// because then it is caused by the closing in the stop
					// method
				} else {
					LOGGER.error("error in client socket", e);
				}
			} catch (IOException ioe) {
				LOGGER.error("IO error in client socket", ioe);
			}
		}
	}

	@Override
	public void stopListening() {
		this.keepListening = false;
		try {
			this.serverSocket.close();
		} catch (IOException e) {
			// ignore failures
		}
	}

	@Override
	public int getPort() {
		return this.serverSocket.getLocalPort();
	}

	/**
	 * marker interface for all server session handler classes
	 * 
	 * @author Sebastian Bader
	 * @author Christian Koch
	 */
	public static interface SessionHandler extends Runnable {
		// empty interface
	}

	/**
	 * a session handler to communicate with the given client socket. all
	 * communication specification is done in this class
	 * 
	 * @author Sebastian Bader
	 * @author Christian Koch
	 */
	protected abstract class AbstractSessionHandler implements SessionHandler {
		protected RequestProcessor<R> requestProcessor;

		protected Socket socket; // client socket
		protected boolean keepAnswering; // communication will be stopped when
											// set to false
		protected OutputStream outstream; // socket output stream (maybe with
											// logging wrapper)
		protected InputStream instream; // socket output stream (maybe with
										// logging wrapper)

		protected String clientHost;
		protected int clientPort;

		public abstract String getConnectionTypeName();

		/**
		 * creates a new session handler for socket connections
		 * 
		 * @param socket
		 *            client socket
		 */
		public AbstractSessionHandler(Socket socket,
				RequestProcessor<R> requestProcessor) {
			this.socket = socket;
			this.keepAnswering = true;

			this.requestProcessor = requestProcessor;
		}

		/**
		 * synchronized, writes given S-Expression to socket output stream
		 * 
		 * @param exp
		 *            S-Expression to be written
		 * @throws IOException
		 *             an error occurred while writing to stream
		 */
		protected void outwrite(Answer answer) throws IOException {
			byte[] array = answer.toByteArray();
			if (array != null) {
				outwrite(array);
				outflush();
			}
		}

		/**
		 * synchronized, writes given string to socket output stream
		 * 
		 * @param s
		 *            String to be written
		 * @throws IOException
		 *             an error occurred while writing to stream
		 */
		protected void outwrite(String s) throws IOException {
			outwrite(Toolbox.toUTF8Bytes(s));
		}

		/**
		 * synchronized, writes given binary data to socket output stream
		 * 
		 * @param data
		 *            data to be written
		 * @throws IOException
		 *             an error occurred while writing to stream
		 */
		protected void outwrite(byte[] data) throws IOException {
			synchronized (this.outstream) {
				this.outstream.write(data);
			}
		}

		/**
		 * synchronized, writes given byte to socket output stream
		 * 
		 * @param data
		 *            byte to be written
		 * @throws IOException
		 *             an error occurred while writing to stream
		 */
		protected void outwrite(int data) throws IOException {
			synchronized (this.outstream) {
				this.outstream.write(data);
			}
		}

		/**
		 * synchronized, flushes socket output stream
		 * 
		 * @throws IOException
		 *             an error occurred while flushing to stream
		 */
		protected void outflush() throws IOException {
			synchronized (this.outstream) {
				this.outstream.flush();
			}
		}

		/**
		 * initialize socket steams, logging and object event handling
		 * 
		 * @throws IOException
		 *             cannot get socket streams
		 */
		protected void init() throws IOException {
			this.clientHost = this.socket.getInetAddress()
					.getCanonicalHostName();
			this.clientPort = this.socket.getPort();

			// init input and output streams
			this.instream = this.socket.getInputStream();
			this.outstream = this.socket.getOutputStream();

			this.requestProcessor.init();

			AbstractServer.this.fireEvent(new ConnectionEvent(
					AbstractServer.ConnectionEvent.Type.CONNECTION_STARTED,
					this.clientHost, this.clientPort, true));
		}

		public abstract R nextRequest();

		@Override
		public void run() {
			try {
				init();

				stdout("Start communication with "
						+ this.socket.getInetAddress() + ":"
						+ this.socket.getPort() + " (type: "
						+ getConnectionTypeName() + ")");

				// waiting for client requests
				while (this.keepAnswering) {
					R request = nextRequest();
					if (request == null)
						this.keepAnswering = false;
					else {
						Answer answer = this.requestProcessor
								.answerRequest(request);
						outwrite(answer);
						if (answer.endSession())
							this.keepAnswering = false;
					}
				}

				// terminating client connection
				// in all cases the socket should be closed afterwards
				if ((this.socket != null) && (!this.socket.isClosed()))
					this.socket.close();
			} catch (SocketException e) {
				if (!this.keepAnswering) {
					// normal connection end exception
				}
			} catch (IOException e) {
				LOGGER.error("IO error of client socket", e);
			} finally {
				stdout("Communication" + " (type: " + getConnectionTypeName()
						+ ") with " + this.socket.getInetAddress() + ":"
						+ this.socket.getPort() + " ended ("
						+ (this.keepAnswering ? "dirty" : "clean") + ")");
				close();
			}
		}

		/**
		 * close connection. Tries it gracefully, kills connection if not
		 * possible.
		 */
		@SuppressWarnings("synthetic-access")
		protected void close() {
			this.requestProcessor.close();

			// close socket gracefully if possible
			if (this.socket != null && !this.socket.isClosed()) {
				try {
					this.socket.close();
				} catch (IOException e) {
					LOGGER.error("error closing client socket", e);
				}
				this.socket = null;
			}
			AbstractServer.this.fireEvent(new ConnectionEvent(
					AbstractServer.ConnectionEvent.Type.CONNECTION_ENDED,
					this.clientHost, this.clientPort, !this.keepAnswering));
		}
	}

}
